package com.atguigu.app.dwd;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.util.DateFormatUtil;
import com.atguigu.util.KafkaUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @Author ysp
 * @Date 2022/12/8 11:37
 */
// 数据流：web/app -> Nginx -> 日志服务器(.log) -> Flume -> Kafka(ODS) -> FlinkApp -> Kafka(DWD)
// 程序:   Mock(lg.sh) -> Flume(f1) -> Kafka(ZK) -> BaseLogApp -> Kafka(ZK)
public class BaseLogApp {

    public static void main(String[] args) throws Exception {
        // TODO 1. 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 1.1 开启CheckPoint
        env.enableCheckpointing(5 * 60000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(10 * 60000L);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5000L));
        // 1.2 设置状态后端
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://192.168.60.202:8020/ck");
        System.setProperty("HADOOP_USER_NAME", "atguigu");

        // TODO 2. 消费Kafka topic_Log 主题的数据创建流
        String topic = "topic_Log";
        String groupId = "base_log_app";

        DataStreamSource<String> kafkaDS = env.addSource(KafkaUtil.getFlinkKafkaConsumer(topic, groupId));

        // TODO 3. 过滤掉非JSON格式的数据 && 将每行数据转换为JSON对象
        OutputTag<String> dirtyTag = new OutputTag<String>("Dirty");
        SingleOutputStreamOperator<JSONObject> cleanStream = kafkaDS.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject jsonObject = JSONObject.parseObject(value);
                    out.collect(jsonObject);
                } catch (Exception e) {
                    ctx.output(dirtyTag, value);
                }
            }
        });

        // TODO 4. 按照Mid分组
        KeyedStream<JSONObject, String> keyedStream = cleanStream.keyBy(data -> data.getJSONObject("common").getString("mid"));

        // TODO 5. 使用状态编程做新老访客标记校验
        SingleOutputStreamOperator<JSONObject> fixedStream = keyedStream.map(new RichMapFunction<JSONObject, JSONObject>() {

            private ValueState<String> lastViewState;

            @Override
            public void open(Configuration parameters) throws Exception {
                lastViewState = getRuntimeContext().getState(
                        new ValueStateDescriptor<String>("lastViewState", String.class)
                );
            }

            // {"common":{"ar":"440000","ba":"iPhone","ch":"Appstore","is_new":"1","md":"iPhone Xs Max","mid":"mid_51315","os":"iOS 13.2.3","uid":"603","vc":"v2.1.132"},"start":{"entry":"notice","loading_time":1087,"open_ad_id":1,"open_ad_ms":9832,"open_ad_skip_ms":0},"ts":1651303983000}
            @Override
            public JSONObject map(JSONObject value) throws Exception {
                // 获取is_new和ts
                String is_new = value.getJSONObject("common").getString("is_new");
                Long ts = value.getLong("ts");
                String curData = DateFormatUtil.toDate(ts);

                // 获取状态中的日期
                String lastData = lastViewState.value();
                // is_new为1
                if ("1".equals(is_new)) {
                    if (lastData == null) {
                        lastViewState.update(curData);
                    } else if (!lastData.equals(curData)) {
                        value.getJSONObject("common").put("is_new", "0");
                    }
                }
                // is_new不为1
                else {
                    lastViewState.update(DateFormatUtil.toDate(ts - 24 * 60 * 60 * 1000L));
                }
                return value;
            }
        });

        // TODO 6. 使用侧输出流进行分流处理   页面日志放到主流  启动、曝光、动作、错误
        // 6.1 定义启动、曝光、动作、错误侧输出流
        OutputTag<String> startTag = new OutputTag<>("startTag");
        OutputTag<String> displayTag = new OutputTag<>("displayTag");
        OutputTag<String> actionTag = new OutputTag<>("actionTag");
        OutputTag<String> errorTag = new OutputTag<>("errorTag");

        // 6.2 分流
        SingleOutputStreamOperator<String> pageDS = fixedStream.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject value, Context ctx, Collector<String> out) throws Exception {
                // 收集错误数据
                JSONObject err = value.getJSONObject("err");
                if (err != null) {
                    ctx.output(errorTag, value.toJSONString());
                }
                // 剔除err字段
                value.remove("err");

                // 收集启动数据
                JSONObject start = value.getJSONObject("start");
                if (start != null) {
                    ctx.output(startTag, value.toJSONString());
                } else {
                    // 获取 ”page“字段
                    JSONObject page = value.getJSONObject("page");
                    // 获取"common"
                    JSONObject common = value.getJSONObject("common");
                    // 获取”ts“
                    Long ts = value.getLong("ts");

                    // 收集曝光数据
                    JSONArray displays = value.getJSONArray("displays");
                    if (displays != null) {
                        for (int i = 0; i < displays.size(); i++) {
                            JSONObject display = displays.getJSONObject(i);
                            JSONObject displayObj = new JSONObject();
                            displayObj.put("display", display);
                            displayObj.put("common", common);
                            displayObj.put("page", page);
                            displayObj.put("ts", ts);
                            ctx.output(displayTag, displayObj.toJSONString());
                        }
                    }
                    // 收集动作数据
                    JSONArray actions = value.getJSONArray("actions");
                    if (actions != null) {
                        for (int i = 0; i < actions.size(); i++) {
                            JSONObject action = actions.getJSONObject(i);
                            JSONObject actionObj = new JSONObject();
                            actionObj.put("action", action);
                            actionObj.put("common", common);
                            actionObj.put("page", page);
                            ctx.output(actionTag, actionObj.toJSONString());
                        }
                    }

                    // 收集页面数据
                    value.remove("displays");
                    value.remove("actions");
                    out.collect(value.toJSONString());
                }
            }
        });

        // TODO 7. 提取各个侧输出流数据
        DataStream<String> startDS = pageDS.getSideOutput(startTag);
        DataStream<String> displayDS = pageDS.getSideOutput(displayTag);
        DataStream<String> actionDS = pageDS.getSideOutput(actionTag);
        DataStream<String> errorDS = pageDS.getSideOutput(errorTag);

        // TODO 8. 将数据打印并写出对应的主题
        pageDS.print("page>>>>>>>>>>");
        startDS.print("start>>>>>>>>>>");
        displayDS.print("display>>>>>>>>>>");
        actionDS.print("action>>>>>>>>>>");
        errorDS.print("error>>>>>>>>>>");

        String page_topic = "dwd_traffic_page_log";
        String start_topic = "dwd_traffic_start_log";
        String display_topic = "dwd_traffic_display_log";
        String action_topic = "dwd_traffic_action_log";
        String error_topic = "dwd_traffic_error_log";

        pageDS.addSink(KafkaUtil.getFlinkKafkaProducer(page_topic));
        startDS.addSink(KafkaUtil.getFlinkKafkaProducer(start_topic));
        displayDS.addSink(KafkaUtil.getFlinkKafkaProducer(display_topic));
        actionDS.addSink(KafkaUtil.getFlinkKafkaProducer(action_topic));
        errorDS.addSink(KafkaUtil.getFlinkKafkaProducer(error_topic));

        // TODO 9. 启动任务
        env.execute("base_log_app");
    }



}
